/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Stack;

import javax.xml.parsers.ParserConfigurationException;
import javax.xml.parsers.SAXParser;
import javax.xml.parsers.SAXParserFactory;

import org.xml.sax.Attributes;
import org.xml.sax.Locator;
import org.xml.sax.SAXException;
import org.xml.sax.SAXParseException;
import org.xml.sax.helpers.DefaultHandler;

/**
 * <code>Cluster</code> represents whole SSS cluster.
 */
@SuppressWarnings("serial")
public class Cluster implements Serializable {
  private static int PORT_MAX = 65535;
  private List<StorageNode> storageNodes = new ArrayList<StorageNode>();
  private Set<InetSocketAddress> workerNodes = new HashSet<InetSocketAddress>();
  // StorageNode, WorkerNode map
  private Map<StorageNode, InetSocketAddress> swMap = new HashMap<StorageNode, InetSocketAddress>();

  /**
   * Exception of error on parsing cluster configuration file.
   */
  public static class ConfigParseException extends SssException {
    private String filename;
    private int lineNumber;
    private int columnNumber;

    private ConfigParseException(String filename, int lineNumber, int columnNumber, String msg, Throwable cause) {
      super(msg, cause);
      this.filename = filename;
      this.lineNumber = lineNumber;
      this.columnNumber = columnNumber;
    }

    private ConfigParseException(String filename, int lineNumber, int columnNumber, Throwable cause) {
      this(filename, lineNumber, columnNumber, (cause != null)?cause.getMessage():null, cause);
    }

    private ConfigParseException(String filename, int lineNumber, int columnNumber, String msg) {
      this(filename, lineNumber, columnNumber, msg, null);
    }

    /**
     * Get filename in which error occurred.
     */
    public String getFileName() {
      return filename;
    }

    /**
     * Get line number in which error occurred.
     */
    public int getLineNumber() {
      return lineNumber;
    }

    /**
     * Get line column number in which error occurred.
     */
    public int getColumnNumber() {
      return columnNumber;
    }

    @Override
    public String toString() {
      StringBuilder buf = new StringBuilder(getClass().getName());
      String message = getLocalizedMessage();
      buf.append("; In ").append(filename);
      buf.append(": line ").append(lineNumber);
      buf.append(", col ").append(columnNumber);

      //append the exception message at the end
      if (message!=null) buf.append(": ").append(message);

      return buf.toString();
    }
  };

  /**
   * Get storage nodes represernted by protocol (memcached, tokyotyrant)
   * and socket address.
   *
   * @return the list of storage node.
   */
  public List<StorageNode> getStorageNodes() {
    return Collections.unmodifiableList(storageNodes);
  }

  /**
   * Get worker nodes represented by Java RMI socket address.
   *
   * @return the set of worker node.
   */
  public Set<InetSocketAddress> getWorkerNodes() {
    return Collections.unmodifiableSet(workerNodes);
  }

  /**
   * Get map of <code>StorageNode</code> and worker node to handle it.
   *
   * @return the map of storage node and worker node.
   */
  public Map<StorageNode, InetSocketAddress> getPairs() {
    return Collections.unmodifiableMap(swMap);
  }

  /**
   * Get worker node to handle the specified <code>StorageNode</code>.
   *
   * @return the worker node to handle th specified <code>StorageNode</code>.
   */
  public InetSocketAddress getWorkerNode(StorageNode sn) {
    return swMap.get(sn);
  }

  /*
   * cluster.conf parser
   *
   * Please refer to README for cluster.conf format
   */

  /**
   * Load cluster configuration from the specified file
   *
   * @param  file  configuration file
   * @return a <code>Cluster</code> object with specified configuration
   * @throws SssException
   */
  public static Cluster loadClusterConf(File file) throws SssException {
    try {
      InputStream is = new FileInputStream(file);
      try {
        return loadClusterConf(file.toString(), is);
      }
      finally {
        is.close();
      }
    } catch (java.io.IOException e) {
      throw new SssException(e);
    }
  }

  /**
   * Load cluster configuration from the specified InputStream.
   *
   * @param  is InputStream to load configuration.
   * @return a <code>Cluster</code> object with specified configuration
   * @throws SssException
   */
  public static Cluster loadClusterConf(InputStream is) throws SssException {
    return loadClusterConf("<no filename>", is);
  }

  /**
   * Load cluster configuration from the specified InputStream.
   *
   * @param  fn file name of InputStream.
   * @param  is InputStream to load configuration.
   * @return a <code>Cluster</code> object with specified configuration
   * @throws SssException
   */
  public static Cluster loadClusterConf(String fn, InputStream is) throws SssException {
    return new ConfigParser(fn).parse(is);
  }

  /**
   * Load cluster configuration from the file cluster.conf in
   * the conf directory
   *
   * @return a <code>Cluster</code> object with default configuration
   * @throws SssException
   */
  public static Cluster loadDefaultConfigFile() throws SssException {
    Cluster cluster = null;

    File clusterConf = new File(Resources.getConfigDirectory(), "cluster.conf");
    cluster = Cluster.loadClusterConf(clusterConf);

    return cluster;
  }

  private static class ConfigParser {
    private String filename;
    private Set<StorageNode> storageNodes = new HashSet<StorageNode>();

    // This saves order of StorageNodes in configuration file.
    private LinkedHashMap<InetSocketAddress, List<StorageNode>> workerNodes =
      new LinkedHashMap<InetSocketAddress, List<StorageNode>>();


    public ConfigParser(String filename) {
      this.filename = filename;
    }

    Cluster parse(InputStream is) throws SssException {
      try {
        // NOTE: This tree does not have attributes.
        Element root = readXML(is);
        if (!root.name.equals("cluster")) {
          parseError(root, "root element is not \"cluster\".");
        }
        parseCluster(root);

        check(storageNodes.size() > 0,  root, "No storageNode.");
        check(workerNodes.size() > 0,  root, "No worker node.");

        return makeCluster();
      } catch (IOException e) {
        throw new SssException(e);
      } catch (ParserConfigurationException e) {
        throw new SssException(e);
      } catch (SAXParseException e) {
        throw new ConfigParseException(filename, e.getLineNumber(), e.getColumnNumber(), e.getMessage(), e);
      } catch (SAXException e) {
        throw new SssException(e);
      }
    }

    private void parseCluster(Element root) throws SssException {
      assert root.name.equals("cluster");

      for (Node n: root.children) {
        if (!(n instanceof Element)) {
          continue;
        }
        Element ch = (Element)n;

        if (ch.name.equals("worker")) {
          parseWorker(ch);
        }
      }
    }

    private Cluster makeCluster() {
      Cluster cluster = new Cluster();

      for (Map.Entry<InetSocketAddress, List<StorageNode>> entry: workerNodes.entrySet()) {
        InetSocketAddress wn = entry.getKey();
        cluster.workerNodes.add(wn);
        for (StorageNode sn: entry.getValue()) {
          cluster.storageNodes.add(sn);
          cluster.swMap.put(sn, wn);
        }
      }

      return cluster;
    }

    private void parseWorker(Element worker) throws SssException {
      assert worker.name.equals("worker");

      Box<String>  host = new Box<String>("host", worker);
      Box<Integer> port = new Box<Integer>("port", worker);
      List<StorageNode> sns = new ArrayList<StorageNode>();

      for (Node n: worker.children) {
        if (!(n instanceof Element)) {
          continue;
        }
        Element ch = (Element)n;
        if (host.is(ch)) { host.set(ch, parseText(ch)); }
        if (port .is(ch)) { port.set(ch, parsePort(ch)); }
        if (ch.name.equals("storage")) {
          sns.add(parseStorage(ch));
        }
      }

      host.appearedCheck();
      port.appearedCheck();
      check(!sns.isEmpty(), worker, "<worker> node has no <storage> nodes.");

      InetSocketAddress w = new InetSocketAddress(host.value, port.value);
      check(!workerNodes.containsKey(w), worker, "The worker node(%s) is already appeared.", w);

      workerNodes.put(w, sns);
    }

    private StorageNode parseStorage(Element storage) throws SssException {
      assert storage.name.equals("storage");

      Box<String>  protocol  = new Box<String>("protocol", storage);
      Box<String>  host      = new Box<String>("host", storage);
      Box<Integer> port      = new Box<Integer>("port", storage);
      Box<String>  dbfactory = new Box<String>("dbfactory", storage);

      for (Node n: storage.children) {
        if (!(n instanceof Element)) {
          continue;
        }
        Element ch = (Element)n;
        if (protocol.is(ch))  { protocol .set(ch, parseText(ch));  }
        if (host.is(ch))      { host     .set(ch, parseText(ch));  }
        if (port.is(ch))      { port     .set(ch, parsePort(ch)); }
        if (dbfactory.is(ch)) { dbfactory.set(ch, parseDB(ch));   }
      }
      protocol.appearedCheck();
      host.appearedCheck();
      port.appearedCheck();
      // dbfactory is optional.

      StorageNode sn = new StorageNode(protocol.value, host.value, port.value, dbfactory.value);
      check(!storageNodes.contains(sn), storage, "The storage node(%s) has already been appeared.", sn);
      storageNodes.add(sn);

      return sn;
    }


    private int parsePort(Element port) throws ConfigParseException {
      String text = parseText(port);
      try {
        int v = Integer.parseInt(text);
        check(0 < v && v <= PORT_MAX, port, "Invalid port number(%d)", v);
        return v;
      } catch (NumberFormatException e) {
        parseError(port, "Invalid port number(%s)", text);
        return 0; // DO NOT REACH HERE.
      }
    }

    private String parseDB(Element ch) throws ConfigParseException {
      String db = parseText(ch);
      try {
        Class<?> clazz = Class.forName(db);
        clazz.getConstructor(); // check has no-argument constructor.

        check(DBFactory.class.isAssignableFrom(clazz), ch,
            "%s does not implement DBFactory.", db);
      } catch (ClassNotFoundException e) {
        parseError(ch, "%s has not found.", db);
      } catch (NoSuchMethodException e) {
        parseError(ch, "%s does not have no-argument constructor.", db);
      }
      return db;
    }

    private String parseText(Element el) throws ConfigParseException {
      check(!el.children.isEmpty(), el, "expect that <%s> has text, but empty.", el.name);
      check(el.children.size() == 1 && (el.children.get(0) instanceof Text), el,
          "expect that <%s> has text only, but <%s> has element(s).", el.name, el.name);
      return ((Text)el.children.get(0)).text.trim();
    }

    private void check(boolean cond, Element el, String fmt, Object... args) throws ConfigParseException {
      if (!cond) {
        parseError(el, fmt, args);
      }
    }

    private void parseError(Element el, String fmt, Object... args) throws ConfigParseException {
      throw new ConfigParseException(filename, el.line, el.col, String.format(fmt, args));
    }

    private class Box<T> {
      public T value = null;
      public Element parent;
      public String name;

      public Box(String name, Element parent) {
        this.name = name;
        this.parent = parent;
      }

      public boolean is(Element ch) {
        return name.equals(ch.name);
      }

      public void set(Element ch, T v) throws ConfigParseException {
        check(value == null, ch,
            "<%s> node has already appeared in this <%s> node.",
            name, parent.name);
        value = v;
      }

      public void appearedCheck() throws ConfigParseException {
        check(value != null, parent,
            "<%s> node does not has <%s> node.",
            parent.name, name);
      }
    }

    // Node of XML tree .
    private static class Node {
    };
    private static class Element extends Node {
      public final String name;
      public final List<Node> children;
      public final int line;
      public final int col;
      public Element(String name, int line, int col) {
        this.name = name;
        this.children = new ArrayList<Node>();
        this.line = line;
        this.col = col;
      }
    };
    private static class Text extends Node {
      final String text;
      public Text(String s) {
        this.text = s;
      }
    };

    /**
     * read xml with Locator.
     *
     * NOTE: This code is copy from the following.
     * - http://eyalsch.wordpress.com/2010/11/30/xml-dom-2/
     */
    private Element readXML(InputStream is) throws IOException, SAXException, ParserConfigurationException {
      SAXParserFactory factory = SAXParserFactory.newInstance();
      SAXParser parser = factory.newSAXParser();

      final Stack<Element> elementStack = new Stack<Element>();
      final StringBuilder textBuffer = new StringBuilder();
      final List<Element> root = new ArrayList<Element>();
      DefaultHandler handler = new DefaultHandler() {
        private Locator locator;

        @Override
        public void setDocumentLocator(Locator locator) {
          this.locator = locator;
        }

        @Override
        public void startElement(String uri, String localName, String qName, Attributes attributes) throws SAXException {
          addTextIfNeeded();
          elementStack.push(new Element(qName, locator.getLineNumber(), locator.getColumnNumber()));
          // WARNING: Ignore Attributes, because attributes are not used.
        }

        @Override
        public void endElement(String uri, String localName, String qName){
          addTextIfNeeded();

          assert elementStack.size() > 0;
          Element el = elementStack.pop();
          if (elementStack.isEmpty()) {
            root.add(el);
          } else {
            elementStack.peek().children.add(el);
          }
        }

        @Override
        public void characters (char ch[], int start, int length) throws SAXException {
          textBuffer.append(ch, start, length);
        }

        // Outputs text accumulated under the current node
        private void addTextIfNeeded() {
          if (textBuffer.length() > 0) {
            elementStack.peek().children.add(new Text(textBuffer.toString()));

            // Clear
            textBuffer.delete(0, textBuffer.length());
          }
        }
      };
      parser.parse(is, handler);
      assert root.size() == 1;

      return root.get(0);
    }
  }
}
